package com.study.spark.scala.nginx

import com.study.spark.scala.nginx.`case`.UrlTopNStat
import com.study.spark.scala.nginx.dao.StatDAO
import com.study.spark.scala.nginx.util.{DateUtils, IPUtils, LogConvertUtil, UrlUtils}
import org.apache.hadoop.hive.ql.parse.WindowingSpec.WindowSpec
import org.apache.spark.sql.{DataFrame, Row, SaveMode, SparkSession}
import org.apache.spark.sql.functions._
import org.apache.spark.sql.types.{StringType, StructField}

import scala.collection.mutable.ListBuffer

/**
  * 统计nginx日志
  * 日志样例：
  * 223.73.60.115 - - [03/Aug/2017:00:00:10 +0800] "GET /styles/css/all.min.css HTTP/1.1" 200 14435 "http://rsks.class.com.cn/help.html?entityName=kehufuwu" "Mozilla/5.0 (Windows NT 6.3; WOW64; Trident/7.0; Touch; rv:11.0) like Gecko" "-"
  *
  * @author: stephen.shen
  * @create: 2019-03-08 17:05
  */
object NginxLogApp {

  def main(args: Array[String]): Unit = {

    val date = "2017-08-03"

    //val path = "D:\\WorkDir\\15出版集团\\web-server\\nginx-logs\\access.log.1818.bak"
    //val path = "D:\\WorkDir\\15出版集团\\web-server\\nginx-logs\\access.log.bak"
    val path = "D:\\WorkDir\\15出版集团\\web-app-server03\\nginx\\logs\\access.log"

    val topn = 10

    val spark = SparkSession.builder()
      .master("NginxLogApp")
      .master("local[*]")
      .getOrCreate()
    val accessRDD = spark.sparkContext.textFile(path)
    // accessRDD.take(10).foreach(println)
    // 数据清洗以及转换成DataFrame
    val accessDF = spark.createDataFrame(accessRDD.map(line => LogConvertUtil.convert(line)).filter(_.length != 0), LogConvertUtil.struct)
    // 按天统计
    //urlTopNStat(spark, accessDF, date, topn)

    // 存储清洗后的数据
    //    accessDF.coalesce(1)
    //      .write
    //      .mode(SaveMode.Overwrite)
    //      //.partitionBy("url")
    //      .csv("D:\\WorkDir\\15出版集团\\web-server\\nginx-logs\\access.log.1818")

    find404Url(spark,accessDF)

    spark.stop()
  }

  // 找出404页面
  def find404Url(spark: SparkSession, df: DataFrame) = {
    import spark.implicits._
    val result = df.filter($"statusCode" === 404)
      .filter($"url".endsWith(".png") or $"url".endsWith(".jpg") or $"url".endsWith(".gif"))
      .select($"url").distinct().collect()
    println(result.toBuffer)
  }


  /**
    * 以html结尾的URL TOP N
    *
    * @param spark
    * @param df
    */
  def urlTopNStat(spark: SparkSession, df: DataFrame, date: String, topn: Int): Unit = {
    //方式一：以SQL的方式进行统计
    val sql = s"select date, url, times, " +
      s"ROW_NUMBER() OVER (PARTITION BY date ORDER BY times DESC) as rank " +
      s"from (select date, url, count(1) as times " +
      s"from logs where date = '$date' and url like '%.html' group by date, url order by times desc) limit $topn"
    df.createOrReplaceTempView("logs")
    val result = spark
      .sql(sql)

    //        方式二：以DataFrame方式进行统计
    //    import spark.implicits._
    //    import org.apache.spark.sql._
    //    val result = df.filter($"date" === date)
    //      // 去掉不是以.html结尾的数据
    //      .filter($"url".endsWith(".html"))
    //      .groupBy("date", "url")
    //      .agg(count("url").as("times"))
    //      .orderBy($"times".desc)
    //      .limit(topn)

    // 方式三：以DataSet方式进行统计
    //    import spark.implicits._
    //    val ds = df.as[Log]
    //    val result = ds.filter(col("date") === date)
    //      .groupBy(col("date"), col("url"))
    //      .agg(count("url").as("times"))
    //      .orderBy(col("times") desc)

    //result.show(false)

    //结果保存
    try {
      result.foreachPartition(partitionOfRecords => {
        // 结果暂存到可变List中
        val list = new ListBuffer[UrlTopNStat]
        partitionOfRecords.foreach(record => {
          list.append(UrlTopNStat(
            record.getAs[String]("date"),
            record.getAs[String]("url"),
            record.getAs[Long]("times")
          ))
        })
        // 结果存储到RDMS中
        StatDAO.insertDayUrlAccessTopN(list)
      })
    } catch {
      case e: Exception => e.printStackTrace()
    }
  }

  case class Log(ip: String, province: String, date: String, time: String, url: String, statusCode: Int, traffic: Long, referer: String)

}
